package com.smsc.headend.task.engine.service.impl;

import cn.hutool.core.date.SystemClock;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import com.smsc.headend.module.data.entity.RdMeterReadsVp;
import com.smsc.headend.task.engine.mapper.RdMeterReadsVpMapper;
import com.smsc.headend.task.engine.service.RdMeterReadsVpService;
import lombok.extern.slf4j.Slf4j;

import org.apache.shardingsphere.driver.jdbc.core.datasource.ShardingSphereDataSource;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service;
import org.springframework.util.CollectionUtils;

import javax.sql.DataSource;
import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.util.List;

/**
 * (业务实现)
 *
 * @author wjh
 * @date 2022/3/21 14:02.
 */
@Service
@Slf4j
public class RdMeterReadsVpServiceImpl extends ServiceImpl<RdMeterReadsVpMapper, RdMeterReadsVp> implements RdMeterReadsVpService {


    @Autowired
    DataSource dataSource;

    @Value("${spring.datasource.dbnames}")
    String[] dnNames;

    @Value("${spring.datasource.dbtype:oracle}")
    String dbType;


    @Override
    public void batchSaveRd(List<RdMeterReadsVp> rdMeterReadsVps) {
        if(!CollectionUtils.isEmpty(rdMeterReadsVps)) {
            if(dbType.equals("oracle")) {
                mergeReads(rdMeterReadsVps);
            } else if (dbType.equals("postgresql")) {
                mergeReadsForPgsql(rdMeterReadsVps);
            } else {
                log.error("batch save rd for dcu ,unknown dbType");
            }
        }
    }


    private Boolean mergeReads(List<RdMeterReadsVp> readsVps) {
        long start = SystemClock.now();
        StringBuilder stringBuilder = new StringBuilder();
        stringBuilder.append("MERGE INTO RD_METER_READS_VP A")
                .append("    USING (SELECT ? AS METER_ID, ? AS SRC, ? AS TV, ? AS SUB_DATA_ITEM_ID, ? AS VP_TEMPLATE_ID, ? AS VALUE, ? AS TV_UPDATE,? AS TASK_NO FROM DUAL) B ON ( A.VP_TEMPLATE_ID=B.VP_TEMPLATE_ID AND A.SUB_DATA_ITEM_ID=B.SUB_DATA_ITEM_ID AND A.TASK_NO=B.TASK_NO)")
                .append("    WHEN MATCHED THEN UPDATE SET A.VALUE=B.VALUE, A.SRC=B.SRC, A.TV=B.TV, A.TV_UPDATE=B.TV_UPDATE ")
                .append("    WHEN NOT MATCHED THEN INSERT(METER_ID, SRC, TV, SUB_DATA_ITEM_ID, VP_TEMPLATE_ID, VALUE, TASK_NO, TV_UPDATE)")
                .append("    VALUES(B.METER_ID, B.SRC, B.TV, B.SUB_DATA_ITEM_ID, B.VP_TEMPLATE_ID, B.VALUE, B.TASK_NO, B.TV_UPDATE)");

        String sql = stringBuilder.toString();
        DataSource defaultDbSource = ((ShardingSphereDataSource) dataSource).getContextManager().getDataSourceMap(((ShardingSphereDataSource) dataSource).getSchemaName()).get(dnNames[0]);

        log.debug(sql);
        try (Connection conn = defaultDbSource.getConnection(); PreparedStatement ps = conn.prepareStatement(sql)) {
            conn.setAutoCommit(false);
            for (RdMeterReadsVp batchAddMeterRead : readsVps) {
                ps.setLong(1, batchAddMeterRead.getMeterId());
                ps.setInt(2, batchAddMeterRead.getSrc());
                ps.setLong(3, batchAddMeterRead.getTv());
                ps.setLong(4, batchAddMeterRead.getSubDataItemId());
                ps.setLong(5, batchAddMeterRead.getVpTemplateId());
                ps.setString(6, batchAddMeterRead.getValue());
                ps.setLong(7, batchAddMeterRead.getTvUpdate());
                ps.setString(8, batchAddMeterRead.getTaskNo());
                ps.addBatch();
            }
            ps.executeBatch();
            conn.commit();
            log.debug("save meter vp records size:{},cost: {}ms", readsVps.size(), SystemClock.now() - start);
            ps.close();
            conn.close();
        } catch (SQLException throwables) {
            log.error("batch insert meter vp records error", throwables);
            return false;
        }

        return true;
    }


    private Boolean mergeReadsForPgsql(List<RdMeterReadsVp> readsVps) {
        long start = SystemClock.now();
        StringBuilder stringBuilder = new StringBuilder();

        stringBuilder.append("with upsert as ")
                .append(" (    update RD_METER_READS_VP A ")
                .append(" SET VALUE=B.VALUE, SRC=B.SRC, TV=B.TV, TV_UPDATE=B.TV_UPDATE ")
                .append("from (SELECT ? AS METER_ID, ? AS SRC, ? AS TV, ? AS SUB_DATA_ITEM_ID, ? AS VP_TEMPLATE_ID, ? AS VALUE, ? AS TV_UPDATE,? AS TASK_NO) B ")
                .append(" where (A.VP_TEMPLATE_ID=B.VP_TEMPLATE_ID AND A.SUB_DATA_ITEM_ID=B.SUB_DATA_ITEM_ID AND A.TASK_NO=B.TASK_NO) ")
                .append("returning A.VP_TEMPLATE_ID,A.SUB_DATA_ITEM_ID,A.TASK_NO ) ")
                .append(" insert into RD_METER_READS_VP(METER_ID,SRC, TV, SUB_DATA_ITEM_ID,VP_TEMPLATE_ID,VALUE,TV_UPDATE,TASK_NO) ")
                .append("select * ")
                .append("from (SELECT ? AS METER_ID, ? AS SRC, ? AS TV, ? AS SUB_DATA_ITEM_ID, ? AS VP_TEMPLATE_ID, ? AS VALUE, ? AS TV_UPDATE,? AS TASK_NO) t ")
                .append("where not exists ( ")
                .append("select 1 from upsert B where t.VP_TEMPLATE_ID=B.VP_TEMPLATE_ID AND t.SUB_DATA_ITEM_ID=B.SUB_DATA_ITEM_ID AND t.TASK_NO=B.TASK_NO );");


        String sql = stringBuilder.toString();
        DataSource defaultDbSource = ((ShardingSphereDataSource) dataSource).getContextManager().getDataSourceMap(((ShardingSphereDataSource) dataSource).getSchemaName()).get(dnNames[0]);

        log.debug(sql);
        try (Connection conn = defaultDbSource.getConnection(); PreparedStatement ps = conn.prepareStatement(sql)) {
            conn.setAutoCommit(false);
            for (RdMeterReadsVp batchAddMeterRead : readsVps) {
                ps.setLong(1, batchAddMeterRead.getMeterId());
                ps.setInt(2, batchAddMeterRead.getSrc());
                ps.setLong(3, batchAddMeterRead.getTv());
                ps.setLong(4, batchAddMeterRead.getSubDataItemId());
                ps.setLong(5, batchAddMeterRead.getVpTemplateId());
                ps.setString(6, batchAddMeterRead.getValue());
                ps.setLong(7, batchAddMeterRead.getTvUpdate());
                ps.setString(8, batchAddMeterRead.getTaskNo());
                ps.setLong(9, batchAddMeterRead.getMeterId());
                ps.setInt(10, batchAddMeterRead.getSrc());
                ps.setLong(11, batchAddMeterRead.getTv());
                ps.setLong(12, batchAddMeterRead.getSubDataItemId());
                ps.setLong(13, batchAddMeterRead.getVpTemplateId());
                ps.setString(14, batchAddMeterRead.getValue());
                ps.setLong(15, batchAddMeterRead.getTvUpdate());
                ps.setString(16, batchAddMeterRead.getTaskNo());
                ps.addBatch();
            }
            ps.executeBatch();
            conn.commit();
            log.debug("save meter vp records size:{},cost: {}ms", readsVps.size(), SystemClock.now() - start);
            ps.close();
            conn.close();
        } catch (SQLException throwables) {
            log.error("batch insert meter vp records error", throwables);
            return false;
        }

        return true;
    }
}
